package com.bw.app.dws1;


import com.alibaba.fastjson.JSONObject;
import com.bw.app.bean1.DwsPopularityMetricbean;
import com.bw.app.bean1.DwsUserBehaviorbean;
import com.bw.app.bean1.ShopPopularityKeySelector;
import com.bw.func.PopularityProcessFunc;
import com.bw.func.RedisSink;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


/**
 * Flink店铺多维度人气指标实时计算
 */
public class DwsFlinkShopPopularityCalculator {
    public class BehaviorWithKey {
        public DwsUserBehaviorbean behavior;
        public Tuple4<String,String,String,String> key;

        public BehaviorWithKey(DwsUserBehaviorbean b,
                               Tuple4<String,String,String,String> k) {
            this.behavior = b;
            this.key = k;
        }
    }

    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2. 配置Kafka消费者
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_action_log", "traffic_action_2212A"));

        // 4. 解析JSON数据为DwsUserBehaviorbean对象
        SingleOutputStreamOperator<DwsUserBehaviorbean> dwsBehaviorbean = source.flatMap(new FlatMapFunction<String, DwsUserBehaviorbean>() {
            @Override
            public void flatMap(String s, Collector<DwsUserBehaviorbean> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                Long ts = jsonObject.getLong("ts");
                String user_id = jsonObject.getJSONObject("common").getString("uid");
                String sku_id = null;
                // 2. 仅在满足条件时给 sku_id 赋值
                if (jsonObject.getJSONObject("action") != null // 先判空，避免空指针异常
                        && "sku_id".equals(jsonObject.getJSONObject("action").getString("item_type"))) {
                    sku_id = jsonObject.getJSONObject("action").getString("item");
                }
                String behavior_type = jsonObject.getJSONObject("action").getString("action_id");
                String ch = jsonObject.getJSONObject("common").getString("ch");
                String shop_id = jsonObject.getJSONObject("page").getString("shop_id");
                DwsUserBehaviorbean dwsUserBehaviorbean = new DwsUserBehaviorbean(ts, user_id, sku_id, behavior_type, ch, shop_id);
                collector.collect(dwsUserBehaviorbean);
            }
        }).filter(t->t.getSku_id()!= null);
        dwsBehaviorbean.print();
//        DataStream<DwsUserBehaviorbean> behaviorStream = source
//               .map(t -> {
//                    JSONObject json = new JSONObject(t);
//                    return new DwsUserBehaviorbean(
//                            json.getLong("timestamp"),
//                            json.getString("userId"),
//                            json.getString("productId"),
//                            json.getString("behaviorType"),
//                            json.getString("channel"),
//                            json.getString("storeId")
//                    );
//                })
//                // 过滤只保留需要的渠道
//               .filter(behavior -> "taobao_search".equals(behavior.getChannel())
//                        || "direct_bus".equals(behavior.getChannel()));
//
//        // 5. 多维度分组并计算人气指标（5分钟滚动窗口）
        DataStream<DwsPopularityMetricbean> popularityStream = dwsBehaviorbean
                .keyBy(new ShopPopularityKeySelector())
                .process(new PopularityProcessFunc());   // ✅ 换这里
        // 6. 将计算结果写入Redis

        popularityStream.addSink(new RedisSink());

        // 7. 输出到控制台用于调试
        popularityStream.print("人气指标: ");

        // 执行Flink作业
        env.execute("DwsFlinkShopPopularityCalculator");
    }
}
